package com.mjk.common.core.dcc;


import com.mjk.common.base.error.BizException;
import com.mjk.common.tools.logger.BizLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@Slf4j
public class DccSessionHolder implements Watcher {
	private final Object waiter = new Object();

	private final CountDownLatch countDownLatch = new CountDownLatch(1);

	private final String zkHostPort;

	private final int connectTimeout;

	private ZooKeeper zk;

	private final List<HoldListener> listeners;

	public DccSessionHolder(String zkHostPort, int connectTimeout) {
		this.zkHostPort = zkHostPort;
		this.connectTimeout = connectTimeout;
		listeners = new ArrayList<>();
	}

	public void addHoldListener(HoldListener listener) {
		this.listeners.add(listener);
	}

	private void notifyReconnectListener(ZooKeeper zk) {
		for (HoldListener listener : this.listeners) {
			try {
				listener.onReconnect(zk);
			} catch (Exception e) {
				log.error("notify holdListener error", e);
			}
		}
	}

	@Override
	public void process(WatchedEvent event) {
		if (event.getType() != Event.EventType.None) {
            return;
        }
        synchronized (waiter) {
            switch (event.getState()) {
            case SyncConnected:
                // zk连接建立成功,或者重连成功
                countDownLatch.countDown();
                waiter.notifyAll();
                log.info("Connected...");
                break;
            case Expired:
                // session过期,这是个非常严重的问题,有可能client端出现了问题,也有可能zk环境故障
                // 此处仅仅是重新实例化zk client
            	log.error("Expired(重连)...");
                zk = null;
                ZooKeeper tmp = connect();
                notifyReconnectListener(tmp);
                break;
            case Disconnected:
            	log.error("链接断开，或session迁移....");
                break;
            case AuthFailed:
                close();
                throw new BizException("ZK Connection auth failed...");
            default:
                break;
            }
        }
	}

	
	public synchronized ZooKeeper connect() {
		try {
			if (zk == null) {
				zk = new ZooKeeper(zkHostPort, connectTimeout, this);
				countDownLatch.await();
			}
		} catch (Exception e) {
			BizLogger.error("zk connection error ,"+e.getMessage(),e);
			Thread.currentThread().interrupt();
		}
		return this.zk;
	}
	
	public void del(String path){
		try{
			zk.delete(path, -1);
		}catch(Exception e){
			log.error("del node error,"+e.getMessage(),e);
			Thread.currentThread().interrupt();
		}
	}
	
	public void close() {
		try {
			synchronized (waiter) {
				if (zk != null) {
					zk.close();
				}
				waiter.notifyAll();
			}
		} catch (InterruptedException e) {
			log.error("release connection error ," + e.getMessage(), e);
			Thread.currentThread().interrupt();
		}
	}

	interface HoldListener {

		void onReconnect(ZooKeeper zk);
	}
}
